package com.lagou.hbase.processor;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.CollectionUtils;

import java.io.IOException;
import java.util.List;

public class FriendProcessor extends BaseRegionObserver {

    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        //获取删除表的数据
        final HTableInterface friendship = e.getEnvironment().getTable(TableName.valueOf("friendship"));
        //获取删除所有列的cells
        List<Cell> cells = delete.getFamilyCellMap().get(Bytes.toBytes("friends"));
        if (CollectionUtils.isEmpty(cells)) {
            friendship.close();
            return;
        }
        //获取删除的列作为rowkey
        Cell cell = cells.get(0);
        //创建Delete进行删除cell对应rowkey的关系
        Delete delete1 = new Delete(CellUtil.cloneQualifier(cell));
        delete1.addColumns(Bytes.toBytes("friends"), CellUtil.cloneRow(cell));
        friendship.delete(delete1);
        //关闭table
        friendship.close();
    }
}
